package studio.raptor.extractor.segment.task;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.LockSupport;

import javax.jms.JMSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import studio.raptor.databus.client.DatabusConfig;
import studio.raptor.databus.client.DatabusConsumer;
import studio.raptor.databus.client.DatabusConsumerBuilder;
import studio.raptor.databus.client.EventMessage;
import studio.raptor.databus.command.*;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.service.EventMetaService;
import studio.raptor.extractor.segment.service.RedisCacheService;

/**
 * Created by liuyong8 on 2017/7/20.
 */
@Component
@Scope("prototype")
public class EventMQExtractingTask implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(EventMQExtractingTask.class);

  private ServerConfig serverConfig;
  private BlockingQueue<EventMeta> eventQueue;
  private int count;
  DatabusConsumer consumer;

  @Autowired
  private EventMetaService eventMetaService;

  @Autowired
  private RedisCacheService redisCacheService;

  @Override
  public void run() {
    try {
      consumer = createConsumer();
    }catch (Exception e){
      e.printStackTrace();
    }

    while (!Thread.currentThread().isInterrupted() && serverConfig.getExtractorMQEnabled().get()) {
      EventMessage eventMessage = null;
      try {
        // TODO 需要将获取超时时间支持配置
        eventMessage = consumer.receiveUnACKMessage(1 * 1000);
        if (null == eventMessage) {
          printPerfTip();
          // 休眠0.5秒
          LockSupport.parkUntil(System.currentTimeMillis() + 5 * 100);
        } else {
          Event event = eventMessage.getEvent();
          if (log.isDebugEnabled()) {
            log.debug("have extracted event {}",event);
          }
          EventMetaData eventMetaData = event.getEventMetaData();
          EventBasicData eventBasicData = event.getEventBasicData();
          EventBusinessData eventBusinessData = event.getEventBusinessData();
          // step1 : 将event 入数据库
          Long metaId = eventMetaService.addEventMetaData(eventBasicData, eventMetaData);

          //step2: 插入到队列
          EventMeta eventMeta = cover2EventMeta(metaId,eventMetaData,eventBasicData,eventBusinessData);
          eventQueue.put(eventMeta);
          eventMetaService.modifyEventMetaStatus2P(eventMeta);

          //step3: 受理报文缓存redis
          redisCacheService.saveEventBody(eventMetaData.getBusinessNo(),serverConfig.getKindId(),eventBusinessData.getContent());

        }
      } catch (Exception e) {
        e.printStackTrace();
      } finally{
        try {
          // ack
          eventMessage.acknowledge();
        } catch (JMSException e) {
          e.printStackTrace();
        }
      }
    }


  }

  private void printPerfTip() {
//    if ((++count) % 1000 == 0) {
      log.warn("mq have not event!");
      count = 0;
//    }
  }

  private EventMeta cover2EventMeta(Long metaId,EventMetaData eventMetaData,EventBasicData eventBasicData,EventBusinessData eventBusinessData){
    EventMeta eventMeta = new EventMeta();
    eventMeta.setMetaId(metaId);

    eventMeta.setEventKindId(eventBasicData.getKindId());
    eventMeta.setDeliverTime(eventBasicData.getDeliverTime());

    eventMeta.setCustId(eventMetaData.getCustId());
    eventMeta.setAcctId(eventMetaData.getAcctId());
    eventMeta.setProdInstId(eventMetaData.getProdInstId());
    eventMeta.setOfferInstId(eventMetaData.getOfferInstId());
    eventMeta.setObjectKey(eventMetaData.getBusinessNo());
    eventMeta.setEventType(0);//普通业务
    eventMeta.setOfferIds(eventMetaData.getOfferIds());
    eventMeta.setOfferProdId(eventMetaData.getOfferProdId());
    eventMeta.setOfferSpecAttr(eventMetaData.getOfferSpecAttrs());
    eventMeta.setOrderTypeId(eventMetaData.getOrderTypeIds());
    eventMeta.setProdIds(eventMetaData.getProdIds());
    eventMeta.setRegionId(eventMetaData.getRegionId());
    eventMeta.setServiceOfferIds(eventMetaData.getServiceOfferIds());
    eventMeta.setServiceSpecAttr(eventMetaData.getServiceSpecAttrs());

    eventMeta.setContent(eventBusinessData.getContent());
    return eventMeta;
  }

  private DatabusConsumer createConsumer() throws Exception {
    DatabusConfig config = new DatabusConfig(serverConfig.getBizName());
    DatabusConsumerBuilder consumerBuilder = new DatabusConsumerBuilder(config);
    DatabusConsumer consumer =
        consumerBuilder.createClientACK(serverConfig.getEventName(), serverConfig.getKindId());
    return consumer;
  }

  public ServerConfig getServerConfig() {
    return serverConfig;
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }

  public void setEventQueue(BlockingQueue<EventMeta> eventQueue) {
    this.eventQueue = eventQueue;
  }
}
